package com.hopu.user.time;

import com.alibaba.fastjson.JSONObject;
import com.hopu.bean.time.TimeTop10Shop;
import com.hopu.util.HBaseUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import scala.Tuple2;

import java.io.IOException;
import java.util.List;

public class TimeTop10ShopAna {
    public static void main(String[] args) {

        SparkConf conf = new SparkConf().setAppName("timeTop10Shop").setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);
        //加载日志文件
        JavaRDD<String> rdd = context.textFile("D:/word/user_session.log");

        //获取日志文件中的hour、cid、name
        JavaPairRDD<TimeTop10Shop, Integer> pairRDD = rdd.mapToPair(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String hour = json.getJSONObject("odate").getString("hour");
            String type = json.getString("event_type");
            String sid = json.getString("shop_id");
            String sname = json.getString("shop_name");

            return new Tuple2<>(new TimeTop10Shop(hour, type, sid, sname), 1);
        });

        //过滤出每个时间段的RDD
        for (int i = 0; i < 24; i++) {
            transAndPut(pairRDD, i+"");
        }

    }

    public static void transAndPut(JavaPairRDD<TimeTop10Shop, Integer> pairRDD, String hour) {
        JavaPairRDD<TimeTop10Shop, Integer> hourRDD = pairRDD.filter(t -> hour.equals(t._1.getHour()));

        //过滤出每个时间段不同操作类型的RDD
        JavaPairRDD<TimeTop10Shop, Integer> viewRDD = hourRDD.filter(t -> "view".equals(t._1.getType()));
        JavaPairRDD<TimeTop10Shop, Integer> cartRDD = hourRDD.filter(t -> "cart".equals(t._1.getType()));
        JavaPairRDD<TimeTop10Shop, Integer> parchaseRDD = hourRDD.filter(t -> "parchase".equals(t._1.getType()));

        //分组统计某个时间段的每个操作类型的根据商品的ID进行分组
        viewRDD = groupBySid(viewRDD);
        cartRDD = groupBySid(cartRDD);
        parchaseRDD = groupBySid(parchaseRDD);

        //合并
        JavaPairRDD<TimeTop10Shop, Integer> result = combinRDD(viewRDD, cartRDD, parchaseRDD);
        //排序
        result = result.sortByKey();
        //取top10
        List<TimeTop10Shop> top10Shops = result.map(t -> t._1).take(10);
        top10Shops.forEach(t -> System.out.println(t));

        //保存到数据库
        putToHBase(top10Shops);

    }

    public static void putToHBase(List<TimeTop10Shop> top10Shops) {
        String[] columns = {"hour", "sid", "sname", "viewCount", "cartCount", "parchaseCount"};
        for (TimeTop10Shop top10Shop: top10Shops) {
            String[] values = {top10Shop.getHour(), top10Shop.getSid(), top10Shop.getSname(),
                    top10Shop.getViewCount()+"", top10Shop.getCartCount()+"", top10Shop.getParchaseCount()+""};
            try {
                HBaseUtils.putOneRowToHbase("time", top10Shop.getHour(), "top10shop", columns, values);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


    public static JavaPairRDD<TimeTop10Shop, Integer> combinRDD(JavaPairRDD<TimeTop10Shop, Integer> viewRDD,
                                 JavaPairRDD<TimeTop10Shop, Integer> cartRDD,
                                 JavaPairRDD<TimeTop10Shop, Integer> parchaseRDD) {

        JavaPairRDD<TimeTop10Shop, Tuple2<Tuple2<Integer, Optional<Integer>>, Optional<Integer>>> resultRDD = viewRDD.leftOuterJoin(cartRDD).leftOuterJoin(parchaseRDD);
        return resultRDD.mapToPair(t -> {
            TimeTop10Shop top10Shop = t._1;
            int viewCount = t._2._1._1;
            int cartCount = t._2._1._2.get();
            int parchaseCount = t._2._2.get();
            top10Shop.setCartCount(cartCount);
            top10Shop.setParchaseCount(parchaseCount);
            return new Tuple2<>(top10Shop, viewCount+cartCount+parchaseCount);
        });
    }

    public static JavaPairRDD<TimeTop10Shop, Integer> groupBySid(JavaPairRDD<TimeTop10Shop, Integer> rdd) {
        JavaPairRDD<TimeTop10Shop, Iterable<Integer>> groupBySid = rdd.groupByKey();

        rdd = groupBySid.mapToPair(t -> {
           TimeTop10Shop top10Shop = t._1;
           int count = 0;
           for (int v: t._2) {
               count += v;
           }
           if (top10Shop.getType().equals("view")) {
               top10Shop.setViewCount(count);
           } else if (top10Shop.getType().equals("cart")) {
               top10Shop.setCartCount(count);
           } else {
               top10Shop.setParchaseCount(count);
           }
           return new Tuple2<>(top10Shop, count);
        });
        return rdd;
    }

}
